package com.primeton.gateway.plugin.mqueue.productor;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author: YunTao.Li
 * @create: 2020-03-15 19:54
 * @description:
 **/
@Component
@EnableScheduling
public class KafkaSender {
    @Autowired
    public KafkaTemplate kafkaTemplate;

    private Logger log = LoggerFactory.getLogger(KafkaSender.class);



    public void sendMessage(String topic, String key, String data) {
        // step1 : 同步发送，同时处理发送结果
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, data);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("kafka sendMessage error, ex = {}, topic = {}, data = {}", ex, topic, data);
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("kafka sendMessage success topic = {}, data = {}", topic, data);
            }
        });
    }
}
